// -*-mode:c++; coding:utf-8-*-

#ifndef _XBASE_MULTI_ROW_MUTATOR_ON_TABLET_HPP_
#define _XBASE_MULTI_ROW_MUTATOR_ON_TABLET_HPP_

#include <vector>

#include <boost/function.hpp>
#include <boost/intrusive_ptr.hpp>
#include <boost/system/error_code.hpp>
#include <boost/asio/deadline_timer.hpp>

#include "session.hpp"
#include "protocol.hpp"
#include "client_options.hpp"
#include "async_module_base.hpp"
#include "row_mutation_data.hpp"

namespace xbase
{
	class multi_row_mutator_on_tablet
		: public async_module_base<io_service_pool::xbase_level, io_service_pool::xbase_level>
	{
	public:
		typedef async_module_base<io_service_pool::xbase_level, io_service_pool::xbase_level> base_type;
		typedef multi_row_mutator_on_tablet this_type;
		typedef boost::intrusive_ptr<this_type> ptr_type;

		typedef session::ptr_type session_ptr;

		typedef std::vector<row_mutation_data_ptr> container_type;
		typedef container_type::const_iterator container_iter;

		typedef boost::system::error_code error_code;

		typedef boost::function<void(std::size_t, const error_code&)> handler_type;

		typedef protocol::packet_type<protocol::BATCH_WRITE_REQ>::type batch_write_req;
		typedef protocol::packet_type<protocol::BATCH_WRITE_REQ>::ptr_type batch_write_req_ptr;
		
		typedef protocol::packet_type<protocol::BATCH_WRITE_RESP>::type batch_write_resp;
		typedef protocol::packet_type<protocol::BATCH_WRITE_RESP>::ptr_type batch_write_resp_ptr;

	public:
		multi_row_mutator_on_tablet(io_service_pool &pool)
			: base_type(pool, "multi_row_mutator_on_tablet")
			, m_options(pool.options())
			, m_retry_times(0)
			, m_success_row(0) {}

		// 向服务器会话(session)上的子表(tablet_name)提交修改
		// (mutation)，时间戳将会作为未指定时间戳的记录的时间戳。
		inline
		void async_apply(session_ptr session,
				 const std::string &tablet_name,
				 const container_type &mutations,
				 uint64_t timestamp,
				 handler_type handler) {
			async_apply(session, tablet_name,
				    mutations.begin(), mutations.end(),
				    timestamp, handler);
		}

		inline
		void async_apply(session_ptr session,
				 const std::string &tablet_name,
				 container_iter start,
				 container_iter end,
				 uint64_t timestamp,
				 handler_type handler) {
			batch_write_req_ptr req_ptr(new batch_write_req());
			protocol::batch_write_request_v2 &body = req_ptr->body();
			body.m_tablet_name = tablet_name;
			for(container_iter it = start; it != end; ++ it)
			{
				body.m_rows.push_back(protocol::row_data_v2(*it, timestamp));
				//(*it)->fill_record_vector(body.m_rows.back().m_records, timestamp);
			}
			async_apply(session, req_ptr, handler);
		}

		inline
		void async_apply(session_ptr session,
				 const batch_write_req_ptr &req_ptr,
				 handler_type handler) {
			m_session = session;
			m_req_ptr = req_ptr;
			m_handler = handler;
			
			m_retry_times = 0; // 重置，这样这个对象可以重用
			m_success_row = 0;
			
			on_async_apply();
		}

	protected:
		// 在 lower_level_service 上执行，除第一次异步提交外
		// （为优化，直接由上层调用，而不再转到自己的
		// lower_level_service 上调用）
		//
		// 否则，在 async_apply 中应该是这样调用 on_async_apply
		// this->get_lower_level_service().dispatch(boost::bind(&this_type::on_async_apply,
		// 						        this_ptr()));
		void on_async_apply();
		// 以下函数总在 lower_level_service 上执行

		void on_response(protocol::packet_ptr ptr, const error_code &socket_ec);

		void on_retry(std::size_t success_row, const error_code &ec);
		
		void on_error(std::size_t success_row, const error_code &ec);

		void dispatch(std::size_t success_row, const error_code &ec);

		ptr_type this_ptr() {
			return ptr_type(this);
		}

	private:
		const client_options &m_options;
		boost::shared_ptr<boost::asio::deadline_timer> m_timer;

		session_ptr m_session;
		batch_write_req_ptr m_req_ptr;
		handler_type m_handler;

		std::size_t m_retry_times;
		std::size_t m_success_row;
	};
	
	typedef multi_row_mutator_on_tablet::ptr_type multi_row_mutator_on_tablet_ptr;
	
	inline
	void async_apply_on_tablet(io_service_pool &pool,
				   session_ptr session,
				   const std::string &tablet_name,
				   const multi_row_mutator_on_tablet::container_type &mutations,
				   uint64_t timestamp,
				   multi_row_mutator_on_tablet::handler_type handler)
	{
		multi_row_mutator_on_tablet_ptr mutator(new_async_module(multi_row_mutator_on_tablet, pool));
		mutator->async_apply(session, tablet_name, mutations, timestamp, handler);
	}

} // namespace xbase

#endif	// _XBASE_MULTI_ROW_MUTATOR_ON_TABLET_HPP_
